package com.orangeandbronze.demos.mapreduce;

import java.util.Arrays;

import com.google.appengine.tools.mapreduce.KeyValue;
import com.google.appengine.tools.mapreduce.Mapper;

public class PhrasesMapper extends Mapper<KeyValue<String, byte[]>, String, String> {

	private static final long serialVersionUID = 5001923576575442541L;

	private static final int PHRASE_LENGTH = 4;

	/*
	 * Based on org.apache.commons.lang.StringUtils.
	 */
	private String join(String s[], String separator) {
		StringBuilder buffer = new StringBuilder();
		if (s.length > 0) {
			buffer.append(s[0]); // first
		}
		int i = 1;
		while (i < s.length) {
			buffer.append(separator);
			buffer.append(s[i++]);
		}
		return buffer.toString();
	}

	@Override
	public void map(KeyValue<String, byte[]> data) {
		final String filename = data.getKey();
		final String line = new String(data.getValue())
				.replaceAll("\\s+", " ") // compact consecutive white spaces
				.replaceAll("[\\.\\?\\!]\\s+", "\n"); // split into sentences
				; 
		final String[] sentences = line.split("\n");
		for (String sentence : sentences) {
			final String[] words = sentence.split("\\W+"); // split into words
			if (words.length < PHRASE_LENGTH) {
				getContext().emit(join(words, ":"), filename);
			} else {
				for (int i = 0; i <= (words.length - PHRASE_LENGTH); i++) {
					getContext().emit(
							join(Arrays.copyOfRange(words, i, i + PHRASE_LENGTH), ":"),
							filename);
				}
			}
		}
	}

	/*
PHRASE_LENGTH = 4

def split_into_sentences(s):
  """Split text into list of sentences."""
  s = re.sub(r"\s+", " ", s)
  s = re.sub(r"[\\.\\?\\!]", "\n", s)
  return s.split("\n")

def phrases_map(data):
  """Phrases demo map function."""
  (entry, text_fn) = data
  text = text_fn()
  filename = entry.filename

  logging.debug("Got %s", filename)
  for s in split_into_sentences(text):
    words = split_into_words(s.lower())
    if len(words) < PHRASE_LENGTH:
      yield (":".join(words), filename)
      continue
    for i in range(0, len(words) - PHRASE_LENGTH):
      yield (":".join(words[i:i+PHRASE_LENGTH]), filename)

  	 */
}
